package org.leo.dm.component.reader.db;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.leo.dm.common.bean.StepResultBean;
import org.leo.dm.common.exception.DpException;
import org.leo.dm.common.exception.DpExceptionCodeEnum;
import org.leo.dm.common.exception.DpExceptionTracker;
import org.leo.dm.component.reader.Reader;
import org.leo.dm.component.util.DbBean;
import org.leo.dm.component.util.DbManager;
import org.leo.dm.util.ToUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * Created on 2016年2月18日
 * <p>
 * Description db方式的读组件
 * <p>
 * Copyright neusoft (c) 2015 .
 * 
 * @author xiao_h
 *
 */
public class DbReader extends Reader {
	Logger log = LoggerFactory.getLogger(this.getClass());

	public DbReader() {
		super();
		// 初始化相关插件
	}

	/**
	 * 用于搬运表的最初声明数据量
	 */
	private Map<String, Integer> declareTableCount = new HashMap<String, Integer>();

	protected String[] sqls;
	protected Connection conn;
	protected String[] sourceTableNames;
	private Map<String, Connection> dbConHolder = new HashMap<String, Connection>();

	private Map<String, DbBean> tableDbInfo;

	public void setTableDbInfo(Map<String, DbBean> tableDbInfo) {
		this.tableDbInfo = tableDbInfo;
	}

	public void setSourceTableNames(String[] sourceTableNames) {
		this.sourceTableNames = sourceTableNames;
	}

	public void setSqls(String[] sqls) {
		this.sqls = sqls;
	}

	public void setConn(Connection conn) {
		this.conn = conn;
	}

	public Map<String, Integer> getDeclareTableCount() {
		return declareTableCount;
	}

	public void setDeclareTableCount(Map<String, Integer> declareTableCount) {
		this.declareTableCount = declareTableCount;
	}

	@Override
	public StepResultBean call() {
		StepResultBean resultBean = new StepResultBean();
		PreparedStatement pstmt = null;
		ResultSet rs = null;
		DbBean dbBean = null;
		try {
			/*
			 * 暂留内容 if (isInitBlockFlag()) { throw new DpException(DpExceptionCodeEnum.COMPOMENT_ERROR, "初始化组件错误"); }
			 */
			storage.putStartQueue(queuePlugin.handle(storage.getStorageName()), null);

			// 获得额外增加数据元名称的个数
			int extendDataNamesLength = extendDataNames.length;

			List<String> temp = null;
			Connection connTemp = null;
			int sqlslength = sqls.length;
			ResultSetMetaData rsmd = null;
			int columnCount = 0;
			int finalColumnCount = 0;
			for (int si = 0; si < sqlslength; si++) {
				try {

					seqmentcounter = 0;// 初始化新段计数器

					// 初始化数据库连接开始
					if (!dbConHolder.containsKey(sourceTableNames[si])) {

						if (tableDbInfo != null && tableDbInfo.containsKey(sourceTableNames[si])) {
							dbBean = tableDbInfo.get(sourceTableNames[si]);
							connTemp = DbManager.getConection(dbBean);

							dbConHolder.put(sourceTableNames[si], connTemp);
						} else {
							dbConHolder.put(sourceTableNames[si], this.conn);
						}
					}
					this.conn = dbConHolder.get(sourceTableNames[si]);
					// 初始化数据库连接结束

					// 获取需要搬运的该表的数据量
					getCount(this.conn, sourceTableNames[si], sqls[si]);

					pstmt = conn.prepareStatement(sqls[si]);
					rs = pstmt.executeQuery(); // 执行预处理语句获取查询结果集
					rsmd = rs.getMetaData();
					columnCount = rsmd.getColumnCount();
					finalColumnCount = columnCount;// 初始化基础字段个数

					temp = new ArrayList<String>();
					for (int i = 0; i < columnCount; i++) {
						if (excludeDataNames != null) {
							if (Arrays.asList(excludeDataNames).contains(rsmd.getColumnLabel(i + 1))) {
								finalColumnCount--;
								continue;
							}
						}
						temp.add(rsmd.getColumnLabel(i + 1));
					}

					dataNames = temp.toArray(new String[temp.size()]);
					for (int i = finalColumnCount; i < finalColumnCount + extendDataNamesLength; i++) {
						dataNames[i] = extendDataNames[i - finalColumnCount];
					}

					if (declareDataNames == null) {
						storage.putStartSeqment(si, segmentPlugin.handle(sourceTableNames[si]), dataNames, null);
					} else {
						storage.putStartSeqment(si, segmentPlugin.handle(sourceTableNames[si]), declareDataNames, null);
					}
					// storage.putStartSeqment(si, segmentPlugin.handle(sourceTableNames[si]), a,null);
					// declareDataNames = null;//使用完后置空

					while (rs.next()) { // 循环遍历查询结果集
						counter++;// 计数器累加
						seqmentcounter++;// 段计数器累加

						dataValues = new String[finalColumnCount + extendDataNamesLength];
						for (int i = 0; i < finalColumnCount; i++) {
							dataValues[i] = columnPlugin.handle(sourceTableNames[si], dataNames[i], rs.getString(dataNames[i]));
						}
						storage.putData(recordPlugin.handle(sourceTableNames[si], dataNames, dataValues, seqmentcounter));
					}

					storage.putEndSeqment(si);

				} catch (Exception e) {
					log.error(DpExceptionTracker.trace(e));
					throw e;
				} finally {
					if (rs != null) {
						rs.close();
					}
					if (pstmt != null) {
						pstmt.close();
					}
				}
			}

			storage.putEndQueue();

			log.info("数据抽取执行完成");

			Map<String, Object> callBackMap = new HashMap<String, Object>();
			callBackMap.put("counter", counter);
			callBack.handle(callBackMap);

			resultBean.setCode("1");
			resultBean.setName("运行正常完成");
			resultBean.setSuccessCount(counter);
		} catch (Exception e) {
			e.printStackTrace();
			log.error(DpExceptionTracker.trace(e));
			resultBean.setCode("-1");
			resultBean.setName("运行发生异常");
			resultBean.setErrorDesc(e.getMessage());
			try {
				this.getFlow().stop();
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		} finally {
			for (Map.Entry<String, Connection> entry : dbConHolder.entrySet()) {
				if (entry.getValue() != null) {
					try {
						if (entry.getValue() != null && !entry.getValue().isClosed()) {
							entry.getValue().close();
						}
					} catch (Exception e1) {
						e1.printStackTrace();
					}
				}
			}
			counter = 0;// 计数器清零
		}

		return resultBean;
	}

	/**
	 * 
	 * Created on 2015年11月11日 .
	 * 
	 * @author xiao_h 获取记录总数子方法
	 * @throws DpException
	 */
	private void getCount(Connection conn, String tableName, String sqlStr) throws DpException {
		PreparedStatement ps = null;
		ResultSet rsCount = null;
		try {
			ToUtil tu1 = new ToUtil();
			String sqlCount = "select count(1) record_ from (" + sqlStr + ") total";
			log.info("getCount:" + sqlCount);
			ps = conn.prepareStatement(sqlCount);
			rsCount = ps.executeQuery();

			if (rsCount.next()) {
				rowCount = rsCount.getInt("record_");
			}
			// this.storage.setDataCount(rowCount);

			// 赋值声明数据量集合
			declareTableCount.put(tableName, rowCount);
			log.info("getCount count:" + rowCount);
			tu1.takeTimePrex("获取记录数花费时间");
		} catch (Exception e) {
			e.printStackTrace();
			log.error(DpExceptionTracker.trace(e));
			try {
				this.getFlow().stop();
			} catch (Exception e1) {
				e1.printStackTrace();
			}
			throw new DpException(DpExceptionCodeEnum.COMPOMENT_ERROR, "读取组价发生错误");
		} finally {
			try {
				if (rsCount != null) {
					rsCount.close();
				}
				if (ps != null) {
					ps.close();
				}

			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}
}
